package DyingBaby.backend.table;

import DyingBaby.backend.catalog.TableSchema;
import DyingBaby.backend.dm.DataManager;
import DyingBaby.backend.tm.TransactionManager;
import DyingBaby.backend.vm.VersionManager;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.function.Predicate;

/**
 * 存储引擎实现
 * 主要功能：表数据的插入、删除、更新和查询，维护主键索引。
 * 主键索引：操作速度更快。
 * 版本控制器（VersionManager）：实现多版本并发控制，支持事务的 ACID 特性。
 * 多版本并发控制（MVCC）：读写不阻塞，通过数据多版本来实现高性能的并发读写。
 * 逻辑删除：通过 vm.delete() 标记记录为删除，后续事务不可见。
 */
public class PagedTableHeap {
    // 数据存储目录
    private static final String DATA_DIR = "data";
    // 文件后缀：存储所有UID的文件
    private static final String UID_SUFFIX = ".uid"; // one uid per line
    // 文件后缀：存储主键到UID映射的文件
    private static final String PK_SUFFIX = ".pk"; // csv: key,uid per line

    private final TableSchema schema; // 表结构定义
    private final TransactionManager tm; // 事务管理器
    private final DataManager dm; // 数据管理器
    private final VersionManager vm; // 版本管理器
    private final Path uidPath; // UID存储文件路径
    private final Path pkPath; // 主键索引文件路径
    private final Map<String, Long> pkIndex = new HashMap<>(); // 主键到UID的内存索引

    // 构造函数：初始化表存储
    public PagedTableHeap(TableSchema schema, TransactionManager tm, DataManager dm) {
        this.schema = schema;
        this.tm = tm;
        this.dm = dm;
        // 创建版本管理器，用于处理多版本并发控制
        this.vm = VersionManager.newVersionManager(tm, dm);
        try {
            // 确保数据目录存在
            Path dataDir = Paths.get(DATA_DIR);
            if (!Files.exists(dataDir)) Files.createDirectories(dataDir);
            // 构建UID和主键索引文件路径
            this.uidPath = dataDir.resolve(schema.getTableName() + UID_SUFFIX);
            this.pkPath = dataDir.resolve(schema.getTableName() + PK_SUFFIX);
            // 创建文件（如果不存在）
            File f = uidPath.toFile();
            if (!f.exists()) f.createNewFile();
            File pf = pkPath.toFile();
            if (!pf.exists()) pf.createNewFile();
            // 从文件加载主键索引到内存
            loadPkIndex();
        } catch (Exception e) {
            throw new RuntimeException("Init PagedTableHeap failed", e);
        }
    }

    // 插入一条记录
    public synchronized void insert(Map<String,Object> record) {
        // 检查主键唯一性
        String pk = schema.getPrimaryKey();
        if (pk != null) {
            Object pkVal = record.get(pk);
            if (pkVal == null) throw new IllegalArgumentException("PRIMARY KEY value required: " + pk);
            String key = keyString(pkVal);
            if (pkIndex.containsKey(key)) throw new IllegalArgumentException("Duplicate PRIMARY KEY: " + key);
        }
        // 编码记录为字节数组
        byte[] row = RowCodec.encode(schema, record);
        long xid = vm.begin(0);
        try {
            // 通过版本管理器插入数据，返回唯一标识符UID
            long uid = vm.insert(xid, row);
            // 将UID追加到UID文件
            try (BufferedWriter bw = new BufferedWriter(new FileWriter(uidPath.toFile(), true))) {
                bw.write(Long.toString(uid));
                bw.newLine();
            }
            // 更新主键索引
            if (pk != null) {
                String key = keyString(record.get(pk));
                pkIndex.put(key, uid);
                persistPkIndex(); // 持久化主键索引到文件
            }
            vm.commit(xid); // 提交事务
        } catch (Exception e) {
            vm.abort(xid); // 发生异常时回滚事务
            throw new RuntimeException("Insert failed", e);
        }
    }

    // 根据条件删除记录
    public synchronized int deleteIf(Predicate<Map<String,Object>> filter) {
        int cnt = 0;
        try (BufferedReader br = new BufferedReader(new FileReader(uidPath.toFile()))) {
            String line;
            // 遍历所有UID
            while ((line = br.readLine()) != null) {
                if (line.isEmpty()) continue;
                long uid = Long.parseLong(line.trim());
                long xid = vm.begin(0);
                byte[] raw = vm.read(xid, uid);
                if (raw != null) {
                    Map<String,Object> rec = RowCodec.decode(schema, raw);
                    // 检查记录是否满足删除条件
                    if (filter == null || filter.test(rec)) {
                        vm.delete(xid, uid); // 标记删除
                        // 从主键索引中移除
                        String pk = schema.getPrimaryKey();
                        if (pk != null) {
                            String key = keyString(rec.get(pk));
                            pkIndex.remove(key);
                            persistPkIndex();
                        }
                        cnt++;
                    }
                }
                vm.commit(xid);
            }
        } catch (Exception e) {
            throw new RuntimeException("Delete failed", e);
        }
        return cnt;
    }

    // 根据条件更新记录
    public synchronized int updateIf(Predicate<Map<String,Object>> filter, Map<String,Object> set) {
        int cnt = 0;
        try {
            // 先收集所有uid，避免在遍历同一文件时追加写入造成无限循环
            java.util.List<Long> uids = new java.util.ArrayList<>();
            try (BufferedReader br = new BufferedReader(new FileReader(uidPath.toFile()))) {
                String line;
                while ((line = br.readLine()) != null) {
                    if (line.isEmpty()) continue;
                    uids.add(Long.parseLong(line.trim()));
                }
            }

            for (long uid : uids) {
                long xid = vm.begin(0);
                byte[] raw = vm.read(xid, uid);
                if (raw != null) {
                    Map<String,Object> rec = RowCodec.decode(schema, raw);
                    if (filter == null || filter.test(rec)) {
                        String pk = schema.getPrimaryKey();
                        String oldKey = pk != null ? keyString(rec.get(pk)) : null;
                        // 应用更新操作
                        for (Map.Entry<String,Object> e : set.entrySet()) rec.put(e.getKey(), e.getValue());
                        // 如果主键变更，检查唯一性
                        if (pk != null) {
                            String newKey = keyString(rec.get(pk));
                            if (!newKey.equals(oldKey) && pkIndex.containsKey(newKey)) {
                                vm.abort(xid);
                                throw new RuntimeException("Duplicate PRIMARY KEY: " + newKey);
                            }
                        }
                        // 重新编码并插入新记录，删除旧记录
                        byte[] row = RowCodec.encode(schema, rec);
                        vm.delete(xid, uid);
                        long newUid = vm.insert(xid, row);
                        // 记录新UID
                        try (BufferedWriter bw = new BufferedWriter(new FileWriter(uidPath.toFile(), true))) {
                            bw.write(Long.toString(newUid));
                            bw.newLine();
                        }
                        // 更新主键索引
                        if (pk != null) {
                            String newKey = keyString(rec.get(pk));
                            if (oldKey != null) pkIndex.remove(oldKey);
                            pkIndex.put(newKey, newUid);
                            persistPkIndex();
                        }
                        cnt++;
                    }
                }
                vm.commit(xid);
            }
        } catch (Exception e) {
            throw new RuntimeException("Update failed", e);
        }
        return cnt;
    }

    // 扫描记录，支持条件过滤
    public synchronized List<Map<String,Object>> scan(Predicate<Map<String,Object>> filter) {
        List<Map<String,Object>> out = new ArrayList<>();
        try (BufferedReader br = new BufferedReader(new FileReader(uidPath.toFile()))) {
            String line;
            while ((line = br.readLine()) != null) {
                if (line.isEmpty()) continue;
                long uid = Long.parseLong(line.trim());
                long xid = vm.begin(0);
                byte[] raw = vm.read(xid, uid);
                vm.commit(xid);
                if (raw == null) continue; // 跳过已删除或不可见的记录
                Map<String,Object> rec = RowCodec.decode(schema, raw);
                if (filter == null || filter.test(rec)) {
                    out.add(rec);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("Scan failed", e);
        }
        return out;
    }

    // 根据主键查询记录
    public synchronized Map<String,Object> getByPrimaryKey(Object key) {
        if (schema.getPrimaryKey() == null) throw new IllegalStateException("No PRIMARY KEY defined");
        Long uid = pkIndex.get(keyString(key));
        if (uid == null) return null;
        try {
            long xid = vm.begin(0);
            byte[] raw = vm.read(xid, uid);
            vm.commit(xid);
            if (raw == null) return null;
            return RowCodec.decode(schema, raw);
        } catch (Exception e) {
            throw new RuntimeException("Get by PK failed", e);
        }
    }

    // 根据主键删除记录
    public synchronized int deleteByPrimaryKey(Object key) {
        if (schema.getPrimaryKey() == null) throw new IllegalStateException("No PRIMARY KEY defined");
        Long uid = pkIndex.get(keyString(key));
        if (uid == null) return 0;
        try {
            long xid = vm.begin(0);
            byte[] raw = vm.read(xid, uid);
            if (raw != null) {
                vm.delete(xid, uid);
            }
            vm.commit(xid);
            pkIndex.remove(keyString(key));
            persistPkIndex();
            return 1;
        } catch (Exception e) {
            throw new RuntimeException("Delete by PK failed", e);
        }
    }

    // 根据主键更新记录
    public synchronized int updateByPrimaryKey(Object key, Map<String,Object> set) {
        if (schema.getPrimaryKey() == null) throw new IllegalStateException("No PRIMARY KEY defined");
        Long uid = pkIndex.get(keyString(key));
        if (uid == null) return 0;
        try {
            long xid = vm.begin(0);
            byte[] raw = vm.read(xid, uid);
            if (raw == null) { vm.commit(xid); return 0; }
            Map<String,Object> rec = RowCodec.decode(schema, raw);
            String pk = schema.getPrimaryKey();
            String oldKey = keyString(rec.get(pk));
            for (Map.Entry<String,Object> e : set.entrySet()) rec.put(e.getKey(), e.getValue());
            // 检查主键唯一性（如果主键被修改）
            if (pk != null) {
                String newKey = keyString(rec.get(pk));
                if (!newKey.equals(oldKey) && pkIndex.containsKey(newKey)) {
                    vm.abort(xid);
                    throw new RuntimeException("Duplicate PRIMARY KEY: " + newKey);
                }
            }
            byte[] row = RowCodec.encode(schema, rec);
            vm.delete(xid, uid);
            long newUid = vm.insert(xid, row);
            // 记录新UID
            try (BufferedWriter bw = new BufferedWriter(new FileWriter(uidPath.toFile(), true))) {
                bw.write(Long.toString(newUid));
                bw.newLine();
            }
            // 更新主键索引
            pkIndex.remove(oldKey);
            pkIndex.put(keyString(rec.get(pk)), newUid);
            persistPkIndex();
            vm.commit(xid);
            return 1;
        } catch (Exception e) {
            throw new RuntimeException("Update by PK failed", e);
        }
    }

    // 将主键值转换为字符串形式（用于索引键）
    private String keyString(Object v) {
        return String.valueOf(v);
    }

    // 从文件加载主键索引到内存
    private void loadPkIndex() {
        pkIndex.clear();
        try (BufferedReader br = new BufferedReader(new FileReader(pkPath.toFile()))) {
            String line;
            while ((line = br.readLine()) != null) {
                if (line.isEmpty()) continue;
                int c = line.lastIndexOf(',');
                if (c <= 0) continue;
                String key = line.substring(0, c);
                long uid = Long.parseLong(line.substring(c+1));
                pkIndex.put(key, uid);
            }
        } catch (Exception e) {
            throw new RuntimeException("Load PK index failed", e);
        }
    }

    // 将内存中的主键索引持久化到文件
    private void persistPkIndex() {
        try (BufferedWriter bw = new BufferedWriter(new FileWriter(pkPath.toFile(), false))) {
            for (Map.Entry<String,Long> e : pkIndex.entrySet()) {
                bw.write(e.getKey());
                bw.write(',');
                bw.write(Long.toString(e.getValue()));
                bw.newLine();
            }
        } catch (Exception e) {
            throw new RuntimeException("Persist PK index failed", e);
        }
    }
} 